package Kafka.Producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;

import java.util.Properties;

public class DefaultPartitioner {
    // TODO 默认分区器 DefaultPartitioner

    //（1）指明partition的情况下，直接将指明的值作为partition值；例如partition=0，所有数据写入分区0
    //（2）没有指明partition值但有key的情况下，将key的hash值与topic的partition数进行取余得到partition值；
        // 例如：key1的hash值=5， key2的hash值=6 ，topic的partition数=2，
        // 那么key1 对应的value1写入1号分区，key2对应的value2写入0号分区。
    //（3）既没有partition值又没有key值的情况下，Kafka采用Sticky Partition（黏性分区器），会随机选择一个分区，并尽可能一直
    // 使用该分区，待该分区的batch已满或者已完成，Kafka再随机一个分区进行使用（和上一次的分区不同）。
        // 例如：第一次随机选择0号分区，等0号分区当前批次满了（默认16k）或者linger.ms设置的时间到，
        // Kafka再随机一个分区进行使用（如果还是0会继续随机）。


    public static void main(String[] args) throws InterruptedException {

        // 1. 创建kafka生产者的配置对象
        Properties properties = new Properties();

        // 2. 给kafka配置对象添加配置信息
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "master:9092");

        // key,value序列化（必须）：key.serializer，value.serializer
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());

        // 3. 创建kafka生产者对象
        KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

        // 4. 调用send方法,发送消息
        for (int i = 0; i < 5; i++) {
            // 添加回调
            // 该方法在Producer收到ack时调用，为异步调用

            // 指定数据发送到1号分区，key为空（IDEA中ctrl + p查看参数）

//            kafkaProducer.send(new ProducerRecord<>("firstTopic", 1,"","ansentai " + i),

            // 依次指定key值为a,b,f ，数据key的hash值与3个分区求余，分别发往1、2、0

            kafkaProducer.send(new ProducerRecord<>("firstTopic", "a","ansentai " + i),
            (metadata, exception) -> {
                        if (exception == null) {
                            // 没有异常,输出信息到控制台
                            System.out.println("主题：" + metadata.topic() + "->" + "分区：" + metadata.partition());
                        } else {
                            // 出现异常打印
                            exception.printStackTrace();
                        }
                    });

            // 延迟一会会看到数据发往不同分区
            Thread.sleep(2);
        }

        // 5. 关闭资源
        kafkaProducer.close();
    }
}
